package disruptor;


import com.lmax.disruptor.AlertException;
import com.lmax.disruptor.EventProcessor;
import com.lmax.disruptor.ExceptionHandler;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.Sequence;
import com.lmax.disruptor.SequenceBarrier;
import com.lmax.disruptor.Sequencer;
import com.lmax.disruptor.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * <p>A {@link WorkProcessor} wraps a single {@link WorkHandler}, effectively consuming the sequence
 * and ensuring appropriate barriers.</p>
 *
 * <p>Generally, this will be used as part of a {@link WorkerPool}.</p>
 *
 * @param <T> event implementation storing the details for the work to processed.
 */
public final class WorkProcessor<T>
    implements EventProcessor
{
  private final AtomicBoolean running = new AtomicBoolean(false);
  private final Sequence sequence = new Sequence(Sequencer.INITIAL_CURSOR_VALUE);
  private final RingBuffer<T> ringBuffer;
  private final SequenceBarrier sequenceBarrier;
  private final WorkHandler<? super T> workHandler;
  private final ExceptionHandler<? super T> exceptionHandler;
  private final Sequence workSequence;

  private final EventReleaser eventReleaser = new EventReleaser()
  {
    @Override
    public void release()
    {
      sequence.set(Long.MAX_VALUE);
    }
  };

  private final TimeoutHandler timeoutHandler;

  /**
   * Construct a {@link WorkProcessor}.
   *
   * @param ringBuffer       to which events are published.
   * @param sequenceBarrier  on which it is waiting.
   * @param workHandler      is the delegate to which events are dispatched.
   * @param exceptionHandler to be called back when an error occurs
   * @param workSequence     from which to claim the next event to be worked on.  It should always be initialised
   *                         as {@link Sequencer#INITIAL_CURSOR_VALUE}
   */
  public WorkProcessor(
      final RingBuffer<T> ringBuffer,
      final SequenceBarrier sequenceBarrier,
      final WorkHandler<? super T> workHandler,
      final ExceptionHandler<? super T> exceptionHandler,
      final Sequence workSequence)
  {
    this.ringBuffer = ringBuffer;
    this.sequenceBarrier = sequenceBarrier;
    this.workHandler = workHandler;
    this.exceptionHandler = exceptionHandler;
    this.workSequence = workSequence;

    if (this.workHandler instanceof EventReleaseAware)
    {
      ((EventReleaseAware) this.workHandler).setEventReleaser(eventReleaser);
    }

    timeoutHandler = (workHandler instanceof TimeoutHandler) ? (TimeoutHandler) workHandler : null;
  }

  @Override
  public Sequence getSequence()
  {
    return sequence;
  }

  @Override
  public void halt()
  {
    running.set(false);
    sequenceBarrier.alert();
  }

  @Override
  public boolean isRunning()
  {
    return running.get();
  }

  /**
   * It is ok to have another thread re-run this method after a halt().
   *
   * @throws IllegalStateException if this processor is already running
   */
  @Override
  public void run()
  {
    if (!running.compareAndSet(false, true))
    {
      throw new IllegalStateException("Thread is already running");
    }
    sequenceBarrier.clearAlert();

    notifyStart();

    boolean processedSequence = true;
    long cachedAvailableSequence = Long.MIN_VALUE;
    long nextSequence = sequence.get();
    T event = null;
    while (true)
    {
      try
      {
        // if previous sequence was processed - fetch the next sequence and set
        // that we have successfully processed the previous sequence
        // typically, this will be true
        // this prevents the sequence getting too far forward if an exception
        // is thrown from the WorkHandler
        if (processedSequence)
        {
          processedSequence = false;
          do
          {
            nextSequence = workSequence.get() + 1L;
            sequence.set(nextSequence - 1L);
          }
          while (!workSequence.compareAndSet(nextSequence - 1L, nextSequence));
        }

        if (cachedAvailableSequence >= nextSequence)
        {
          event = ringBuffer.get(nextSequence);
          workHandler.onEvent(event);
          processedSequence = true;
        }
        else
        {
          cachedAvailableSequence = sequenceBarrier.waitFor(nextSequence);
        }
      }
      catch (final TimeoutException e)
      {
        notifyTimeout(sequence.get());
      }
      catch (final AlertException ex)
      {
        if (!running.get())
        {
          break;
        }
      }
      catch (final Throwable ex)
      {
        // handle, mark as processed, unless the exception handler threw an exception
        exceptionHandler.handleEventException(ex, nextSequence, event);
        processedSequence = true;
      }
    }

    notifyShutdown();

    running.set(false);
  }

  private void notifyTimeout(final long availableSequence)
  {
    try
    {
      if (timeoutHandler != null)
      {
        timeoutHandler.onTimeout(availableSequence);
      }
    }
    catch (Throwable e)
    {
      exceptionHandler.handleEventException(e, availableSequence, null);
    }
  }

  private void notifyStart()
  {
    if (workHandler instanceof LifecycleAware)
    {
      try
      {
        ((LifecycleAware) workHandler).onStart();
      }
      catch (final Throwable ex)
      {
        exceptionHandler.handleOnStartException(ex);
      }
    }
  }

  private void notifyShutdown()
  {
    if (workHandler instanceof LifecycleAware)
    {
      try
      {
        ((LifecycleAware) workHandler).onShutdown();
      }
      catch (final Throwable ex)
      {
        exceptionHandler.handleOnShutdownException(ex);
      }
    }
  }
}

